package com.gitee.ly;

import com.gitee.ly.model.ServiceInfo;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class DefaultDiscoverAndLoadBalancer implements DiscoverAndLoadBalancer {
    private ClientManage clientManage;
    private final Map<String, AtomicLong> serviceUseCount = new ConcurrentHashMap<>();
    private final ScheduledThreadPoolExecutor autoClearUseCountScheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);

    public DefaultDiscoverAndLoadBalancer(ClientManage clientManage) {
        this.clientManage = clientManage;
        if (!clientManage.isEmpty()) {
            this.autoClearUseCountScheduledThreadPoolExecutor.scheduleWithFixedDelay(() -> autoClearUseCount(this), 1, 10, TimeUnit.MINUTES);
        }
    }

    //隔10分钟清理一次计数
    private void autoClearUseCount(DefaultDiscoverAndLoadBalancer thisLoadBalancer) {
        try {
            List<String> scountKeyList = new ArrayList<>(thisLoadBalancer.serviceUseCount.keySet());
            for (String scountKey : scountKeyList) {
                AtomicLong atomicLong = thisLoadBalancer.serviceUseCount.get(scountKey);
                if (atomicLong != null) {
                    atomicLong.set(0L);
                }
            }
        } catch (Throwable ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    private String getServiceUseCountKey(ServiceInfo serviceInfo) {
        return String.format("%s:%s", serviceInfo.getIp(), serviceInfo.getPort());
    }

    @Override
    public ServiceInfo chooseOne(String serviceName) {
        List<ServiceInfo> canUseServiecList = this.clientManage.findCanUse(serviceName);
        if (canUseServiecList == null || canUseServiecList.isEmpty()) {
            return null;
        } else {
            ServiceInfo useService = canUseServiecList.get(0);
            AtomicLong useCount = serviceUseCount.computeIfAbsent(getServiceUseCountKey(canUseServiecList.get(0)), k -> new AtomicLong(0));
            for (ServiceInfo serviceInfo : canUseServiecList) {
                AtomicLong thisUseCount = serviceUseCount.computeIfAbsent(getServiceUseCountKey(serviceInfo), k -> new AtomicLong(0));
                if (thisUseCount.get() < useCount.get()) {
                    useService = serviceInfo;
                    useCount = thisUseCount;
                }
            }
            //计数清理，防止越界
            long newVal = useCount.incrementAndGet();
            if (newVal >= (Integer.MAX_VALUE - 100)) {
                // 原子清零
                for (ServiceInfo serviceInfo : canUseServiecList) {
                    AtomicLong atomicLong = serviceUseCount.get(getServiceUseCountKey(serviceInfo));
                    if (atomicLong != null) {
                        atomicLong.set(0L);
                    }
                }
                useCount.set(1L);
            }

            return useService;
        }
    }
}
